package com.wxy.demo.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.concurrent.TimeUnit;

/**
 * <p>
 * 订阅
 * </p>
 *
 * @author e-Xiaoyuan.Wang
 * @since 2022/7/25 15:55
 */
@Slf4j
public class SubscribeDemo {

    public static void main(String[] args) throws MqttException, InterruptedException {
        // 连接配置
        String broker = "tcp://localhost:1883";
        String userName = "admin";
        String password = "admin";
        String clientId = "subscribe";
        boolean cleanSession = true;
        // 订阅配置
        String topic = "test";
        int qos = 1;
        // 创建客户端
        MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
        // 创建连接参数
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(cleanSession);
        options.setUserName(userName);
        options.setPassword(password.toCharArray());
        // 设置回调函数
        client.setCallback(new MqttCallback() {

            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("connectionLost---------");
                cause.printStackTrace();
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("topic:" + topic);
                System.out.println("Qos:" + message.getQos());
                System.out.println("isRetained:" + message.isRetained());
                System.out.println("content:" + new String(message.getPayload()));
                System.out.println("---------------");

            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("deliveryComplete---------" + token.isComplete());
            }

        });
        client.connect(options);
        // 订阅消息
        client.subscribe(topic, qos);
        log.info("订阅话题：topic = {},qos = {}", topic, qos);

        TimeUnit.SECONDS.sleep(60);
        // 关闭连接
        client.disconnect();
        client.close();
    }
}
